package com.zookeeper;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * Created by buguniao on 16/5/4.
 */
class ReadLock implements Lock{
    private ZooKeeper zk;
    private String root;
    private Object readCondition;
    private String curPath="";
    public ReadLock(ZooKeeper zk,String root,Object readCondition){
        this.zk=zk;
        this.root=root;
        this.readCondition=readCondition;
    }
    public void lock() {
        try {
            Stat stat = zk.exists(root,false);
            if(stat==null){// 如果节点不存在，返回的就是null
                try {//这个地方必须有catch，否则就没有下面的步骤
                    zk.create(root, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                }catch(KeeperException.NodeExistsException e){
                    System.out.println("节点已经存在，获取锁失败,path="+e.getPath());

                }

            }
            curPath = zk.create(root+"/read-","".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

            boolean flag ;

            while(true){
                flag=true;//true必须放在里面，每次notify后都要重新初始化变量
                List<String> list = zk.getChildren(root,false);
//                System.out.println("r- 排序前:"+list);
                //必须排序，因为返回的是无序状态
                Collections.sort(list, new Comparator<String>() {
                    public int compare(String o1, String o2) {
                        String s1 = o1.substring(o1.indexOf("-")+1);
                        String s2 = o2.substring(o2.indexOf("-")+1);
                        return s1.compareTo(s2);
                    }
                });
//                System.out.println("r- 排序后:"+list);

                int index = list.indexOf(curPath.substring(curPath.lastIndexOf("/")+1));
                for(int i=index;i>=0;i--){
                    String s = list.get(i);
                    if(s.contains("write")){

                        /*zk.exists(root + "/" + s, new Watcher() {
                            public void process(WatchedEvent watchedEvent) {
                                if(watchedEvent.getState()== Event.KeeperState.SyncConnected){
                                    if(watchedEvent.getType()== Event.EventType.NodeDeleted){
                                        synchronized (readCondition){
                                            readCondition.notifyAll();
                                            System.err.println("write notify");
                                        }
                                    }
                                }
                            }
                        });*/
                        zk.exists(root + "/" + s,true);
                        System.out.println(Thread.currentThread() + " 获取readlock， " + curPath + " 等待 " + root + "/" + s + " 释放");
                        flag=false;
                        break;
                    }
                }

                if(!flag){
                    synchronized (readCondition){
                        //这个地方为什么要使用超时wait，因为上面exists方法是基于先检查后操作的，并不是原子操作，有可能注册监听器的时候原来
                        //存在的节点已经没有了，这样时间永远不会触发，wait永远不会被唤醒，那么本来这里是想使用条件队列来代替轮询的低效性
                        //但是这里又回归原始，值得优化
                        readCondition.wait(10000);
                        System.err.println("20s 检查");
                    }
                }else{
                    System.out.println(Thread.currentThread()+" 获取到了read lock,"+curPath);
                    break;
                }
            }

        }catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void lockInterruptibly() throws InterruptedException {

    }

    public boolean tryLock() {
        return false;
    }

    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }

    public void unlock() {
        try {
            zk.delete(curPath,-1);
            System.out.println(Thread.currentThread()+" 释放了read lock,"+curPath);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    public Condition newCondition() {
        return null;
    }
}

class WriteLock implements Lock{
    private ZooKeeper zk;
    private String root;
    private Object writeCondition;
    private String curPath;
    public WriteLock(ZooKeeper zk,String root,Object writeCondition){
        this.zk=zk;
        this.root=root;
        this.writeCondition=writeCondition;
    }

    public void lock() {
        try {
            Stat stat = zk.exists(root,true);
            if(stat==null){
                try{
                    zk.create(root,"".getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
                }catch (KeeperException.NodeExistsException e){
                    System.out.println("节点已经存在，获取锁失败,path="+e.getPath());

                }

            }
            curPath = zk.create(root+"/write-","".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);


            while(true){
                List<String> list = zk.getChildren(root,false);
//                System.out.println("排序前："+list);
                Collections.sort(list, new Comparator<String>() {
                    public int compare(String o1, String o2) {
                        String s1 = o1.substring(o1.indexOf("-")+1);
                        String s2 = o2.substring(o2.indexOf("-")+1);
                        return s1.compareTo(s2);
                    }
                });
//                 System.out.println("排序后："+list);
                if(list!=null && !list.isEmpty()){
                    int index = list.indexOf(curPath.substring(curPath.lastIndexOf("/")+1));

                    if(index==0){
                        System.out.println(Thread.currentThread()+" 获取到了write lock,"+curPath);
                        break;
                    }else{
                        /*zk.exists(root + "/" + list.get(index - 1), new Watcher() {
                            public void process(WatchedEvent watchedEvent) {
                                if(watchedEvent.getState()== Event.KeeperState.SyncConnected){
                                    if(watchedEvent.getType()==Event.EventType.NodeDeleted){
                                        System.err.println("notify:"+watchedEvent);
                                        synchronized (writeCondition){
                                            writeCondition.notifyAll();
                                        }
                                    }
                                }
                            }
                        });*/

                        zk.exists(root + "/" + list.get(index - 1),true);
                        System.out.println(Thread.currentThread() + " 获取writelock， " + curPath + " 等待 " + root + "/" + list.get(index - 1));
                        synchronized (writeCondition){
                            writeCondition.wait(30000);
                            System.err.println("30s 检查");
                        }
                    }
                }else{
                    throw new Exception("xxxx");
                }
            }

        }catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void lockInterruptibly() throws InterruptedException {

    }

    public boolean tryLock() {
        return false;
    }

    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }

    public void unlock() {
        try {
            zk.delete(curPath,-1);
            System.out.println(Thread.currentThread()+" 释放了write lock,"+curPath);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    public Condition newCondition() {
        return null;
    }
}

public class MyReadWriteLock implements Watcher{
    private String root="/locks";
    private CountDownLatch latch = new CountDownLatch(1);
    private Object readCondition = new Object();
    private Object writeCondition= new Object();
    ZooKeeper zk = null;
    public MyReadWriteLock(){
        try {
            zk=new ZooKeeper("localhost:2181",5000,this);
            latch.await();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    public Lock readLock(){
        return new ReadLock(zk,root,readCondition);
    }
    public Lock writeLock(){
        return new WriteLock(zk,root,writeCondition);
    }


    public void process(WatchedEvent event) {
        try{
            if(event.getState()==Event.KeeperState.SyncConnected){
                if(event.getType()== Event.EventType.None){
                    System.out.println("session 建立");
                    latch.countDown();
                }else if(event.getType()== Event.EventType.NodeDeleted){
                    if(event.getPath().contains("write")){
                        synchronized (readCondition){
                            System.err.println("read notify:"+event.getPath());
                            readCondition.notifyAll();//虽然在上面有很多的watcher注册在这个writelock对应的节点上，但是实际上只会有一个
                            //watcher注册上了，因为在client的watchmanager中Map<String,Set<Watcher>>,所以上都是使用的是默认的watcher
                            //来注册，不管注册多少次都只有一个watcher，也就是说只会被触发一次，但这不会造成什么问题，因为在同一个zksession中
                            //使用的是同一个readCondition,只要readCondition被notify一次就可以使得所有的wait得到唤醒
                        }
                    }else{
                        synchronized (writeCondition){
                            System.err.println("write notify:"+event.getPath());
                            writeCondition.notifyAll();
                        }
                    }
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws BrokenBarrierException, InterruptedException {
        final MyReadWriteLock readWriteLock = new MyReadWriteLock();

        ExecutorService pool = Executors.newFixedThreadPool(100);

        for(int i=0;i<100;i++){

            if(i%10!=0){
                pool.submit(new Runnable() {
                    public void run() {
                        Lock readLock = readWriteLock.readLock();
                        readLock.lock();
                        for(int i=0;i<3;i++){
                            try {
                                System.out.println(Thread.currentThread()+" , read lock "+ i);
                                Thread.sleep(1000);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        }

                        readLock.unlock();
                    }
                });
            }else {
                pool.submit(new Runnable() {
                    public void run() {
                        Lock writeLock = readWriteLock.writeLock();
                        writeLock.lock();
                        for(int i=0;i<5;i++){
                            try {
                                System.out.println(Thread.currentThread()+" , write lock "+ i);
                                Thread.sleep(1000);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        }
                        writeLock.unlock();
                    }
                });
            }
        }

        pool.shutdown();
    }
}
